package com.yifeng.repo.flink.data.transport.streaming.functions.task;

import java.sql.Blob;
import java.sql.Clob;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.collections.CollectionUtils;
import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Strings;
import com.yifeng.repo.flink.data.transport.config.DataSourceConfig;
import com.yifeng.repo.flink.data.transport.dto.DataPacket;
import com.yifeng.repo.flink.data.transport.dto.DataRow;
import com.yifeng.repo.flink.data.transport.relational.connection.JdbcConnectionFactory;
import com.yifeng.repo.flink.data.transport.streaming.functions.api.Engine;
import com.yifeng.repo.flink.data.transport.utils.executors.FireExecutor;
import com.yifeng.repo.flink.data.transport.utils.queues.ChangeEventQueue;

import io.debezium.DebeziumException;
import io.debezium.jdbc.JdbcConnection.ConnectionFactory;
import io.debezium.util.VariableLatch;

/**
 * 全量同步的任务抽象处理类
 * @author lijing
 * @param <T>
 * @since 2023-01-03
 *
 */
public abstract class CommonFullSyncTask<T> implements Engine {
	
	private static final Logger LOGGER = LoggerFactory.getLogger(CommonFullSyncTask.class);
    
	protected final DataSourceConfig dataSourceConfig;
	
	private final ConnectionFactory factory;
	
	private ChangeEventQueue<DataPacket> queue;
    
	//主线程是否正在执行
	private volatile AtomicBoolean isRunning = new AtomicBoolean(false);
	
	//子线程是否正在执行
	private volatile boolean subTaskIsRunning = false;
	
    private final VariableLatch latch = new VariableLatch(0);
    
    private static final Duration SHUTDOWN_WAIT_TIMEOUT  = Duration.ofSeconds(15);
    
    public static final int DEFAULT_MAX_QUEUE_SIZE = 100;
    
    private transient FireExecutor fireExecutor;
    
    private final AtomicInteger threadNumber = new AtomicInteger(0);
    
    private static final String DATE_PATTERN = "yyyy-MM-dd HH:mm:ss";

    private static ThreadLocal<SimpleDateFormat> threadLocal = new ThreadLocal<SimpleDateFormat>(){
        @Override
        protected SimpleDateFormat initialValue() {
            return new SimpleDateFormat(DATE_PATTERN);
        }
    };
    
	public CommonFullSyncTask(DataSourceConfig dataSourceConfig) {
		this.dataSourceConfig = dataSourceConfig;
        //实例化阻塞队列
        this.queue = new ChangeEventQueue.Builder<DataPacket>()
                .maxQueueSize(DEFAULT_MAX_QUEUE_SIZE)
                .build();
        if(dataSourceConfig.getParallelism() > 0) {
        	fireExecutor = new FireExecutor(dataSourceConfig);
    	}
		this.factory = new JdbcConnectionFactory(dataSourceConfig);
	}
	
	/**
	 * 一：从数据库中获取数据
	 * 二：将数据推送到下游
	 */
	@Override
	public void run(SourceContext<DataPacket> ctx) {
		if (!isRunning.compareAndSet(false, true)) {
			return;
		}
		latch.countUp();
		subTaskIsRunning = true;
		long startTime = System.currentTimeMillis();
		try {
			//多表情况下循环异步执行
			if(CollectionUtils.isNotEmpty(dataSourceConfig.getTableList())) {
				dataSourceConfig.getTableList().forEach(tableName -> {
					threadNumber.getAndIncrement();
					fireExecutor.submitTask(new Runnable() {
						@Override
						public void run() {
							loadFullData(tableName);
						}
					} );
				});
				poll(ctx);
				LOGGER.info("Finish load full data,  total cost {} MS ",System.currentTimeMillis() - startTime);	
			}
		} catch(Exception e) {
			close();
			LOGGER.error("执行时出现异常：" + e);
		} finally {
			latch.countDown();
			isRunning.set(false);
		}
	}
	
	/**
	   *  同步方法，数据直接返回给下一个tansfermation
	 *  不启用
	 */
	@Deprecated
	@Override
	public void execute(SourceContext<DataPacket> ctx) {
	}
	
	/**
	 * 获取全量数据
	 */
	private void loadFullData(String tableName) {
		long startTime = System.currentTimeMillis();
		//数据库查询
        PreparedStatement ps = null;
        ResultSet resultSet = null;
        Connection connection = null;
		try {
			connection = factory.connect(null);
			//组装查询SQL语句,根据tableName找到对应的列名，然后进行组装，如果没有找到，默认查询全部字段
			ps = connection.prepareStatement(chooseSql(tableName), ResultSet.TYPE_FORWARD_ONLY,ResultSet.CONCUR_READ_ONLY);
			int fetchSize = Integer.valueOf(dataSourceConfig.getFetchSize());
			ps = setFetchDirection(ps,fetchSize);
			resultSet = ps.executeQuery();
			int count = 0;
			int queueNo = 0;
			List<DataRow> dataRowList = new ArrayList<>(fetchSize);
			while (subTaskIsRunning && resultSet.next()) {
				count++;
				DataRow dataRow = getPubRowData(tableName,resultSet);
				dataRowList.add(dataRow);
				if (dataRowList.size() >= fetchSize) {
					DataPacket dataPacket = assembleDataPacket(queueNo++,count,dataRowList,tableName,false);;
					queue.enqueue(dataPacket);
					dataRowList = new ArrayList<>(fetchSize);
				}
			}
			if(subTaskIsRunning) {
				if (dataRowList.size() > 0) {
					DataPacket lastDataPacket = assembleDataPacket(queueNo,count,dataRowList,tableName,false);
					queue.enqueue(lastDataPacket);
				}
				LOGGER.info("fetch all data row,  total cost {} MS  ,The table name is {},The count is {}",System.currentTimeMillis() - startTime, tableName,count);
				DataPacket endDataPacket = assembleDataPacket(0,-count,Collections.singletonList(assembleDataRow(DataRow.Type.end,tableName,null)),tableName,true);
				queue.enqueue(endDataPacket);
			}
		} catch (Exception e) {
			LOGGER.error("出现异常 " + e.getMessage());
			e.printStackTrace();
			//抛出异常
			try {
				if(connection != null) {
				  doClose(connection);
				}
				if(ps != null) {
				  ps.close();
				}
			} catch (SQLException ex) {
				ex.printStackTrace();
			}
		} finally {
			try {
				if(connection != null) {
					LOGGER.info("connection " + connection.isClosed());
					doClose(connection);
					LOGGER.info("connection close");
				}
				if(ps != null) {
					LOGGER.info("ps " + ps.isClosed());
					ps.close();	
					LOGGER.info("ps close");
				}
			} catch (SQLException e) {
				e.printStackTrace();
			}
			threadNumber.getAndDecrement();
			LOGGER.info("线程执行完成");
		}
	
	}
	
	public abstract DataRow assembleDataRow(DataRow.Type type,String tableName,HashMap<String, Object> dataMap);
	
	public abstract DataPacket assembleDataPacket(int queueNo,long idx ,List<DataRow> rows,String tableName,boolean end);
	
	private void doClose(Connection connection) throws SQLException {
        ExecutorService executor = Executors.newSingleThreadExecutor();
        // attempting to close the connection gracefully
        Future<Object> futureClose = executor.submit(() -> {
        	connection.close();
            LOGGER.info("Connection gracefully closed");
            return null;
        });
        try {
            futureClose.get(10, TimeUnit.SECONDS);
        }
        catch (ExecutionException e) {
            if (e.getCause() instanceof SQLException) {
                throw (SQLException) e.getCause();
            }
            else if (e.getCause() instanceof RuntimeException) {
                throw (RuntimeException) e.getCause();
            }
            throw new DebeziumException(e.getCause());
        }
        catch (TimeoutException | InterruptedException e) {
            LOGGER.warn("Failed to close database connection by calling close(), attempting abort()");
            connection.abort(Runnable::run);
        }
        finally {
            executor.shutdownNow();
        }
    }
	
	private String chooseSql(String tableName) {
		if(Objects.isNull(dataSourceConfig.getSqlCondition())) {
			return assembleFieldSql(tableName);
		}
		String udSql = dataSourceConfig.getSqlCondition().get(tableName);
		if(Strings.isNullOrEmpty(udSql)) {
			return assembleFieldSql(tableName);
		}
		return udSql;
	}
	
	/**
	 * 组装全量同步的sql语句
	 * @param tableName 表名
	 * @return 组装后的sql语句
	 */
	private String assembleFieldSql(String tableName) {
		//根据表名找到对应的列名列表
		if(Objects.isNull(dataSourceConfig.getTableFieldMap())) {
			return allFieldSql(tableName);
		}
		return appointFieldSql(tableName,dataSourceConfig.getTableFieldMap().get(tableName));
	}
	
	/**
	 * sql语句中包含所有字段
	 * @param tableName 表名
	 * @return 组装后的sql语句
	 */
	private String allFieldSql(String tableName) {
		return "select * from " + tableName;
	}
	
	/**
	 * sql语句中指定字段
	 * @param tableName 表名
	 * @param tableFieldList 字段名列表
	 * @return 组装后的sql语句
	 */
	private String appointFieldSql(String tableName,List<Object> tableFieldList) {
		if(CollectionUtils.isEmpty(tableFieldList)) {
			return allFieldSql(tableName);
		}
		String sql = "select ";
		for(int index=0;index<tableFieldList.size();index++) {
			if(index == tableFieldList.size() -1) {
				sql = sql + tableFieldList.get(index);
				break;
			} 
			sql = sql + tableFieldList.get(index) + ",";
		}
		sql = sql + " from " + tableName;
		return sql;
	}
	
	/**
	 * 从队列中poll全量数据
	 * @param ctx 上下文
	 */
	private void poll(SourceContext<DataPacket> ctx) {
		try {
			while (isRunning()) {
				DataPacket dataPacket = (DataPacket) queue.poll(500, TimeUnit.MILLISECONDS);
				if(!Objects.isNull(dataPacket)) {
					ctx.collect(dataPacket);
				}
				//判断子线程是否结束，结束后拉取所有数据后推出循环
				if(isDone()) {
					pollAllData(ctx);
					break;
				}
			 }
			//主线程结束后拉取所有数据
			pollAllData(ctx);
		} catch (InterruptedException e) {
			LOGGER.error("从队列中获取数据异常：" + e);
		}
	}
	
	/**
	 * 判断全量获取数据是否完成
	 * @return true: 完成，false：未完成
	 */
	private boolean isDone() {
        return threadNumber.intValue() <= 0;	
	}
	
	/**
	 * 从队列中拉取最后的数据
	 * @param ctx 上下文
	 */
	private void pollAllData(SourceContext<DataPacket> ctx) {
		try {
			// 如果任务结束了，将队列中的所有数据取出后退出循环
			while (true) {
				DataPacket dataPacketLast = (DataPacket) queue.poll();
				if (Objects.isNull(dataPacketLast)) {
					break;
				}
				ctx.collect(dataPacketLast);
			}
		} catch (InterruptedException e) {
			LOGGER.error("从队列中获取数据异常：" + e);
		}
	}
	
	private DataRow getPubRowData(String tableName, ResultSet resultSet) throws SQLException {
        ResultSetMetaData metaData = resultSet.getMetaData();
        int columnCount = metaData.getColumnCount();
        HashMap<String, Object> dataMap = new HashMap<>();
        for(int i = 1; i<= columnCount ; i++){
            String columnName = metaData.getColumnName(i);
            Object value =getResultSetValue(resultSet, i);
            dataMap.put(columnName,value);
        }
        //根据不同的数据类型处理数据
        return assembleDataRow(DataRow.Type.r,tableName,dataMap);
    }
	
	 public Object getResultSetValue(ResultSet rs, int index)  throws SQLException {
	        Object obj = rs.getObject(index);
	        if(Objects.isNull(obj)) {
	        	return obj;
	        }
	        if (obj instanceof Blob) {
	            Blob blob = (Blob) obj;
	            obj = blob.getBytes(1, (int) blob.length());
	            return obj;
	        } 
	        if (obj instanceof Clob) {
	            Clob clob = (Clob) obj;
	            obj = clob.getSubString(1, (int) clob.length());
	            return obj;
	        }
	        if (obj instanceof LocalDateTime){
	        	try {
	        		obj = ((LocalDateTime)obj).format(DateTimeFormatter.ofPattern(DATE_PATTERN));
	        	} catch (Exception e) {
	        		LOGGER.error("数据解析异常1：" + e);
	        	}
	            return obj;
	        }
	        if (obj instanceof java.sql.Date && "java.sql.Timestamp".equals(rs.getMetaData().getColumnClassName(index))) {
	            return rs.getTimestamp(index);
	        } 
	        if (obj instanceof Timestamp){
	        	try {
		             return threadLocal.get().format(obj);
		        } catch (Exception e) {
		            LOGGER.error("数据解析异常2：" + e);
		        }
	            return obj;
	        }
	        String className  = obj.getClass().getName();
	        if ("oracle.sql.TIMESTAMP".equals(className) || "oracle.sql.TIMESTAMPTZ".equals(className)) {
	            obj = rs.getTimestamp(index);
	            return obj;
	        }
	        if (className != null && className.startsWith("oracle.sql.DATE")) {
	            String metaDataClassName = rs.getMetaData().getColumnClassName(index);
	            if ("java.sql.Timestamp".equals(metaDataClassName) || "oracle.sql.TIMESTAMP".equals(metaDataClassName)) {
	                obj = rs.getTimestamp(index);
	            } else {
	                obj = rs.getDate(index);
	            }
	            return obj;
	        }
	        return obj;
	    }
	
    protected  abstract PreparedStatement setFetchDirection(PreparedStatement ps,int fetchSize) throws Exception;
    
    @Override
	public void close() {
    	//先关闭子线程，再关闭主线程
    	subTaskIsRunning = false;
    	//强制停止之前等待10秒钟等线程执行完成
        try {
        	LOGGER.info("等待15秒钟等待线程结束");
        	latch.await(SHUTDOWN_WAIT_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
		} catch (InterruptedException e) {
			//ingore
		}
        //关闭子线程池
        shutdownExecutor();
        //关闭主线程
        isRunning.set(false);
    }
    
	private void shutdownExecutor() {
		//关闭线程池和子线程
		if(fireExecutor != null) {
			LOGGER.info("shutting down executor");
			fireExecutor.shutdownNow();
		}
		//等待阻塞15秒钟
		if(fireExecutor != null) {
			fireExecutor.awaitTermination(SHUTDOWN_WAIT_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);	
		}
	}
    
    public boolean isRunning() {
        return this.isRunning.get();
    }
    
}
